package cn.com.bluemoon.daps.sync.core.sync.compare;

import cn.com.bluemoon.daps.common.sql.IDbDataSource;
import cn.com.bluemoon.daps.model.entity.DapDataModelSubordinateKey;
import cn.com.bluemoon.daps.model.entity.DapDataModelTable;
import cn.com.bluemoon.daps.model.entity.DapDataModelTableField;
import cn.com.bluemoon.daps.model.entity.DapDataModelTablePartition;
import cn.com.bluemoon.daps.sync.base.CompareResult;
import cn.com.bluemoon.daps.sync.base.ModifyType;
import cn.com.bluemoon.daps.sync.base.ModifyType.ModifyLevel;
import cn.com.bluemoon.daps.sync.base.SqlCompareResult;
import cn.com.bluemoon.daps.sync.core.sync.GlobalConfig;
import cn.com.bluemoon.daps.sync.core.sync.TableModel;
import cn.com.bluemoon.daps.sync.core.sync.Tuple2;
import cn.com.bluemoon.daps.sync.core.sync.compare.db2model.Db2ModelFieldComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.db2model.Db2ModelIndexComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.db2model.Db2ModelPartitionComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.db2model.Db2ModelTableComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.devmodel.Dev2ModelFieldComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.devmodel.Dev2ModelIndexComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.devmodel.Dev2ModelPartitionComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.devmodel.Dev2ModelTableComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.sql.SqlFieldComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.sql.SqlIndexComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.sql.SqlPartitionComparator;
import cn.com.bluemoon.daps.sync.core.sync.compare.sql.SqlTableComparator;
import cn.com.bluemoon.daps.sync.vo.applypublish.ModelCompareVo;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;


/**
 * 两个模型（版本）进行对比
 *
 * @author Jarod.Kong
 */
@Slf4j
public class ModelCompareClient {
    private ModelCompareClient() {
        // nothing
    }

    /**
     * 对比，存储
     *
     * @param v1AndV2DbFields 存储集合
     * @param v1              版本1 根Id进行分组
     * @param v2              版本2 根Id进行分组
     * @param <T>             实体对象 有IDbDataSource#getId
     */
    public static <T extends IDbDataSource> void analyze(List<Tuple2<T>> v1AndV2DbFields,
                                                         List<T> v1, List<T> v2) {
        assert v1AndV2DbFields != null;
        assert v1 != null;
        assert v2 != null;
        // 注意：目前是拿两个表、字段、索引、分区的id为关联关系
        Map<String, T> v1Map = v1.stream().collect(Collectors.toMap(T::getBmKey, e -> e, (a, b) -> b));
        Map<String, T> v2Map = v2.stream().collect(Collectors.toMap(T::getBmKey, e -> e, (a, b) -> b));
        v1Map.forEach((k, v) -> {
            // 删除
            if (!v2Map.containsKey(k)) {
                v1AndV2DbFields.add(new Tuple2<>(v, null));
            }
        });
        v2Map.forEach((k, v) -> {
            // 新增
            // 共同的字段
            v1AndV2DbFields.add(new Tuple2<>(v1Map.getOrDefault(k, null), v));
        });
    }

    /**
     * 获取某个列表+排序（存在定义排序字段的）
     *
     * @param tableModel2 集合
     * @param fun         转换
     * @param <T>         数据实例
     * @return 结果集
     */
    private static <T extends IDbDataSource> List<T> listOrEmptyByTableModel(TableModel tableModel2,
                                                                             Function<TableModel, List<T>> fun) {
        return Optional.ofNullable(tableModel2).map(fun)
                .orElseGet(Collections::emptyList).stream()
                .sorted((o1, o2) -> {
                    if (o1 == null || o1.getRank() == null) {
                        return -1;
                    }
                    if (o2 == null || o2.getRank() == null) {
                        return -1;
                    }
                    return o1.getRank().compareTo(o2.getRank());
                }).collect(Collectors.toList());
    }

    /**
     * 合并表信息，根据两个版本的id对应关系，进行组合，若一方没有则存储为null
     *
     * @param modelV1 v1
     * @param modelV2 v2
     * @return 合并表信息
     */
    public static List<TableModel> unionTableModels(List<TableModel> modelV1, List<TableModel> modelV2) {
        List<Tuple2<TableModel>> v1AndV2DbTableModels = Lists.newArrayList();
        // 对比版本-分类两个版本表对应的数据 为一组 tuple2
        analyze(v1AndV2DbTableModels, modelV1, modelV2);
        //采集每个表在两个版本的对应关系 tuple2(第一版，第二版）
        List<TableModel> tableModels = Lists.newArrayList();
        for (Tuple2<TableModel> v1AndV2DbTable : v1AndV2DbTableModels) {
            TableModel tm = new TableModel();
            TableModel tableModel1 = v1AndV2DbTable.getOldT();
            TableModel tableModel2 = v1AndV2DbTable.getNewT();
            // 存储组合两个版本对应的表tuple
            DapDataModelTable oldT = tableModel1 != null ? tableModel1.getTable() : null;
            DapDataModelTable newT = tableModel2 != null ? tableModel2.getTable() : null;
            Tuple2<DapDataModelTable> tableTuple2 = new Tuple2<>(oldT, newT);
            tm.setV1AndV2DbTable(tableTuple2);
            // 组合两个版本的字段tuple
            List<DapDataModelTableField> fieldsV1 = listOrEmptyByTableModel(tableModel1, TableModel::getFields);
            List<DapDataModelTableField> fieldsV2 = listOrEmptyByTableModel(tableModel2, TableModel::getFields);
            // 增加排序后的前字段名称，来标识该字段的前一位字段名 alter table xx change field_name xx after pre_field_name
            normalizeFieldRanks(fieldsV1, fieldsV2);
            analyze(tm.getV1AndV2DbFields(), fieldsV1, fieldsV2);
            // 组合两个版本的索引tuple
            List<DapDataModelSubordinateKey> indexesV1 = listOrEmptyByTableModel(tableModel1, TableModel::getIndexes);
            List<DapDataModelSubordinateKey> indexesV2 = listOrEmptyByTableModel(tableModel2, TableModel::getIndexes);
            analyze(tm.getV1AndV2DbIndexes(), indexesV1, indexesV2);
            // 组合两个版本的分区tuple
            List<DapDataModelTablePartition> partitionsV1 = listOrEmptyByTableModel(tableModel1, TableModel::getPartitions);
            List<DapDataModelTablePartition> partitionsV2 = listOrEmptyByTableModel(tableModel2, TableModel::getPartitions);
            analyze(tm.getV1AndV2DbPartitions(), partitionsV1, partitionsV2);
            tableModels.add(tm);
        }
        log.info("对比组合完成,size={}", v1AndV2DbTableModels.size());
        return tableModels;
    }

    @SafeVarargs
    private static void normalizeFieldRanks(List<DapDataModelTableField>... fields) {
        Optional.ofNullable(fields)
                .filter(lists -> lists.length > 0)
                .ifPresent(fs -> {
                    for (List<DapDataModelTableField> f : fs) {
                        normalizeFieldRank(f);
                    }
                });
    }

    private static void normalizeFieldRank(List<DapDataModelTableField> f) {
        // 一个字段无需分析after
        if (f.size() <= 1) {
            return;
        }
        // 首个无after
        for (int i = 1; i < f.size(); i++) {
            DapDataModelTableField curF = f.get(i);
            DapDataModelTableField preF = f.get(i - 1);
            curF.setAfterFieldName(preF.getFieldName());
        }
    }

    /**
     * 对比入口类
     *
     * @param modelV1 v1 old
     * @param modelV2 v2 new
     * @return 每个表的对比情况
     */
    public static <R> List<TableCompareRes<R>> doCompareWrapper(BizType bizType,
                                                                GlobalConfig globalConfig,
                                                                List<TableModel> modelV1,
                                                                List<TableModel> modelV2) {
        List<Map<ModifyLevel, List<CompareResult<R>>>> res = doCompare(bizType, globalConfig, modelV1, modelV2);
        List<TableCompareRes<R>> tableCompareRes = res.stream().map(m -> {
            List<CompareResult<R>> collect = m.entrySet().stream().flatMap(d -> d.getValue().stream())
                    .collect(Collectors.toList());
            return collect;
        }).map(c -> {
            CompareResult<R> result = c.stream().filter(t -> t.getTableId() != null && t.getTableName() != null).findFirst().orElse(c.get(0));
            return TableCompareRes.<R>builder().compareResults(c).tableInfo(result).globalConfig(globalConfig).build();
        }).collect(Collectors.toList());
        return tableCompareRes;
    }

    /**
     * 去泛型的对比结果
     *
     * @param bizType      对比模式
     * @param globalConfig 全局配置
     * @param modelV1      v1 old
     * @param modelV2      v2 new
     * @return 每个表的对比情况
     */
    public static List<ModelCompareVo> doCompareV2(BizType bizType,
                                                   GlobalConfig globalConfig,
                                                   List<TableModel> modelV1,
                                                   List<TableModel> modelV2) {
        List<Map<ModifyLevel, List<CompareResult<SqlCompareResult>>>> compareR = doCompare(bizType, globalConfig, modelV1, modelV2);
        List<ModelCompareVo> compareVos = compareResultConvertVo(compareR);
        return compareVos;
    }

    /**
     * 比较结果转换
     *
     * @param compareR 对比结果转换
     * @return list
     */
    public static List<ModelCompareVo> compareResultConvertVo(List<Map<ModifyType.ModifyLevel, List<CompareResult<SqlCompareResult>>>> compareR) {
        List<ModelCompareVo> compareVos = compareR.stream().map(c -> {
            List<CompareResult<SqlCompareResult>> tc = c.get(ModifyType.ModifyLevel.TABLE);
            List<CompareResult<SqlCompareResult>> fc = c.get(ModifyType.ModifyLevel.FIELD);
            List<CompareResult<SqlCompareResult>> ic = c.get(ModifyType.ModifyLevel.INDEX);
            List<CompareResult<SqlCompareResult>> pc = c.get(ModifyType.ModifyLevel.PARTITION);
            CompareResult<SqlCompareResult> b = Stream.of(tc, fc, ic, pc).filter(Objects::nonNull).flatMap(Collection::stream).findFirst().orElseGet(() -> null);
            if (b == null) {
                return null;
            }
            return ModelCompareVo.builder()
                    .modelId(b.getModelId()).modelName(b.getModelName())
                    .tableId(b.getTableId()).tableName(b.getTableName())
                    .dbName(b.getDbName())
                    .tableCompares(tc)
                    .fieldCompares(fc)
                    .indexCompares(ic)
                    .partitionCompares(pc).build();
        }).filter(Objects::nonNull).collect(Collectors.toList());
        return compareVos;
    }

    /**
     * 对比入口类
     *
     * @param bizType      对比模式
     * @param globalConfig 全局配置
     * @param modelV1      v1 old
     * @param modelV2      v2 new
     * @return 每个表的对比情况
     */
    @SuppressWarnings("unchecked")
    public static <R> List<Map<ModifyLevel, List<CompareResult<R>>>> doCompare(BizType bizType,
                                                                               GlobalConfig globalConfig,
                                                                               List<TableModel> modelV1,
                                                                               List<TableModel> modelV2) {
        // 分类
        assert modelV1 != null;
        assert modelV2 != null;
        modelV1 = modelV1.stream().filter(t -> t != null && t.getTable() != null).collect(Collectors.toList());
        modelV2 = modelV2.stream().filter(t -> t != null && t.getTable() != null).collect(Collectors.toList());
        // 组合
        List<TableModel> tableModels = unionTableModels(modelV1, modelV2);
        // 对两组表、字段、索引、分区排列好的进行对比，生成sql变化语句
        List<Map<ModifyLevel, List<CompareResult<R>>>> allTableResult = new ArrayList<>(modelV1.size());
        for (TableModel v1AndV2 : tableModels) {
            // 有序
            Map<ModifyLevel, List<CompareResult<R>>> compareResults = Maps.newLinkedHashMap();
            Tuple2<DapDataModelTable> v1AndV2DbTable = v1AndV2.getV1AndV2DbTable();
            // 表对比
            BaseComparator<DapDataModelTable, R> comparatorT = bizType.compareToT(globalConfig, v1AndV2DbTable);
            comparatorT.setV1AndV2DbFields(v1AndV2.getV1AndV2DbFieldsSortByRank());
            comparatorT.setV1AndV2DbIndexes(v1AndV2.getV1AndV2DbIndexes());
            comparatorT.setV1AndV2DbPartitions(v1AndV2.getV1AndV2DbPartitions());
            comparatorT.setModifyLevel(ModifyLevel.TABLE);
            List<CompareResult<R>> compareResultsT = comparatorT.compare(v1AndV2DbTable.getOldT(), v1AndV2DbTable.getNewT());
            // 字段对比
            BaseComparator<DapDataModelTableField, R> comparatorF = bizType.compareToF(globalConfig, v1AndV2DbTable);
            comparatorF.setV1AndV2DbFields(v1AndV2.getV1AndV2DbFieldsSortByRank());
            comparatorF.setV1AndV2DbIndexes(v1AndV2.getV1AndV2DbIndexes());
            comparatorF.setV1AndV2DbPartitions(v1AndV2.getV1AndV2DbPartitions());
            comparatorF.setModifyLevel(ModifyLevel.FIELD);
            List<CompareResult<R>> compareResultsF = comparatorF.compares(v1AndV2.getV1AndV2DbFieldsSortByRank());
            // 索引对比
            BaseComparator<DapDataModelSubordinateKey, R> comparatorI = bizType.compareToI(globalConfig, v1AndV2DbTable);
            comparatorI.setV1AndV2DbFields(v1AndV2.getV1AndV2DbFieldsSortByRank());
            comparatorI.setV1AndV2DbIndexes(v1AndV2.getV1AndV2DbIndexes());
            comparatorI.setV1AndV2DbPartitions(v1AndV2.getV1AndV2DbPartitions());
            comparatorI.setModifyLevel(ModifyLevel.INDEX);
            List<CompareResult<R>> compareResultsI = comparatorI.compares(v1AndV2.getV1AndV2DbIndexes());
            // 分区对比
            BaseComparator<DapDataModelTablePartition, R> comparatorP = bizType.compareToP(globalConfig, v1AndV2DbTable);
            comparatorP.setV1AndV2DbFields(v1AndV2.getV1AndV2DbFieldsSortByRank());
            comparatorP.setV1AndV2DbIndexes(v1AndV2.getV1AndV2DbIndexes());
            comparatorP.setV1AndV2DbPartitions(v1AndV2.getV1AndV2DbPartitions());
            comparatorP.setModifyLevel(ModifyLevel.PARTITION);
            List<CompareResult<R>> compareResultsP = comparatorP.compares(v1AndV2.getV1AndV2DbPartitions());
            if (!compareResultsT.isEmpty()) {
                log.debug("表变化：{}", compareResultsT.size());
                compareResults.put(ModifyLevel.TABLE, compareResultsT);
            }
            if (!compareResultsF.isEmpty()) {
                log.debug("字段变化：{}", compareResultsF.size());
                compareResults.put(ModifyLevel.FIELD, compareResultsF);
            }
            if (!compareResultsI.isEmpty()) {
                log.debug("索引变化：{}", compareResultsI.size());
                compareResults.put(ModifyLevel.INDEX, compareResultsI);
            }
            if (!compareResultsP.isEmpty()) {
                log.debug("分区变化{}", compareResultsP.size());
                compareResults.put(ModifyLevel.PARTITION, compareResultsP);
            }
            if (!compareResults.isEmpty()) {
                allTableResult.add(compareResults);
            }
        }
        return allTableResult;
    }

    @SuppressWarnings("rawtypes")
    public enum BizType {
        SQL {
            @Override
            public BaseComparator compareToT(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new SqlTableComparator(globalConfig, tableTuple2);
            }

            @Override
            public BaseComparator compareToF(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new SqlFieldComparator(globalConfig, tableTuple2);
            }

            @Override
            public BaseComparator compareToI(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new SqlIndexComparator(globalConfig, tableTuple2);
            }

            @Override
            public BaseComparator compareToP(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new SqlPartitionComparator(globalConfig, tableTuple2);
            }
        },
        Dev2Model {
            @Override
            public BaseComparator compareToT(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new Dev2ModelTableComparator(globalConfig, tableTuple2);
            }

            @Override
            public BaseComparator compareToF(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new Dev2ModelFieldComparator(globalConfig, tableTuple2);
            }

            @Override
            public BaseComparator compareToI(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new Dev2ModelIndexComparator(globalConfig, tableTuple2);
            }

            @Override
            public BaseComparator compareToP(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new Dev2ModelPartitionComparator(globalConfig, tableTuple2);
            }
        },
        Db2Model {
            @Override
            public BaseComparator compareToT(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new Db2ModelTableComparator(globalConfig, tableTuple2);
            }

            @Override
            public BaseComparator compareToF(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new Db2ModelFieldComparator(globalConfig, tableTuple2);
            }

            @Override
            public BaseComparator compareToI(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new Db2ModelIndexComparator(globalConfig, tableTuple2);
            }

            @Override
            public BaseComparator compareToP(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2) {
                return new Db2ModelPartitionComparator(globalConfig, tableTuple2);
            }
        };

        public abstract BaseComparator compareToT(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2);

        public abstract BaseComparator compareToF(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2);

        public abstract BaseComparator compareToI(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2);

        public abstract BaseComparator compareToP(GlobalConfig globalConfig, Tuple2<DapDataModelTable> tableTuple2);
    }

}

